package com.amon.amonplayer.rxjava;

import com.amon.amonplayer.rxjava.function.Consumer;
import com.amon.amonplayer.rxjava.function.Function;
import com.amon.amonplayer.rxjava.schedulder.Schedulers;

public abstract class Observable<T> implements ObservableSource<T> {

    public static <T> Observable<T> just(T s) {
        return onAssembly(new ObservableJust<>(s));
    }

    @Override
    public void subscribe(Observer<T> observer) {
        subscribeActual(observer);
    }

    public void subscribe(Consumer<? super T> onNext) {
        subscribe(onNext, null, null);
    }

    private void subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Consumer<? super T> onComplete) {
        subscribe(new LambdaObserver<T>(onNext, onError, onComplete));
    }

    private static <T> Observable<T> onAssembly(Observable<T> source) {
        return source;
    }

    protected abstract void subscribeActual(Observer<T> s);

    public <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
        return onAssembly(new ObservableMap<>(this, mapper));
    }

    public Observable<T> subscribeOn(Schedulers schedulers) {
        return onAssembly(new ObservableSchedulers<>(this, schedulers));
    }

    public Observable<T> observeOn(Schedulers schedulers) {
        return onAssembly(new ObservableObserverOn<>(this, schedulers));
    }
}
